#pragma once
#include "buffer.hpp"
#include <mutex>
#include <atomic>
#include <condition_variable>
#include <functional>
#include <thread>

namespace ns_log
{
    using AsyncFSCallback = std::function<void(FSBuffer &)>;
    using AsyncDBCallback = std::function<void(DBBuffer &)>;

    enum class AsyncType
    {
        ASYNC_SAFE,
        ASYNC_UNSAFE
    };

    class AsyncLooper
    {
        struct Resource
        {
            std::mutex _mutex;
            std::condition_variable _cond_produce;
            std::condition_variable _cond_consume;
        };
    public:
        using ptr = std::shared_ptr<AsyncLooper>;
        AsyncLooper(AsyncFSCallback fs_callback,AsyncDBCallback db_callback, AsyncType async_type = AsyncType::ASYNC_SAFE)
            : _fs_callback(fs_callback),_db_callback(db_callback), _async_type(async_type), _isrunning(true),
            _fs_worker(&AsyncLooper::fs_thread_routine, this),_db_worker(&AsyncLooper::db_thread_routine,this) {}

        ~AsyncLooper()
        {
            if (_isrunning)
            {
                _isrunning = false;
                if (_fs_worker.joinable())
                {
                    _fs_resource._cond_produce.notify_all();
                    _fs_resource._cond_consume.notify_all();
                    _fs_worker.join();
                }
                if(_db_worker.joinable())
                {
                    _db_resource._cond_produce.notify_all();
                    _db_resource._cond_consume.notify_all();
                    _db_worker.join();
                }
            }
        }

        void push_fs(const char *data, size_t len)
        {
            {
                std::unique_lock<std::mutex> ulock(_fs_resource._mutex);
                if (_async_type == AsyncType::ASYNC_SAFE && _fs_buffer_produce.writeableSize() < len)
                {
                    _fs_resource._cond_produce.wait(ulock, [this, len]() -> bool
                                       { return !_isrunning || _fs_buffer_produce.writeableSize() >= len; });
                }
                _fs_buffer_produce.push(data, len);
            }
            // 唤醒消费者
            _fs_resource._cond_consume.notify_all();
        }

        void push_db(const LogMessage& message)
        {
            {
                std::unique_lock<std::mutex> ulock(_db_resource._mutex);
                if (_async_type == AsyncType::ASYNC_SAFE && _db_buffer_produce.full())
                {
                    _db_resource._cond_produce.wait(ulock, [this]() -> bool
                                       { return !_isrunning || !_db_buffer_produce.full(); });
                }
                _db_buffer_produce.push(message);
            }
            _db_resource._cond_consume.notify_all();
        }

    private:
        void fs_thread_routine()
        {
            while (true)
            {
                {
                    // 当停止运行，且生产缓冲区没有数据了，才能退出
                    if (!_isrunning && _fs_buffer_produce.empty())
                        break;
                    std::unique_lock<std::mutex> ulock(_fs_resource._mutex);
                    _fs_resource._cond_consume.wait(ulock, [this]() -> bool
                                       { return !_isrunning || !_fs_buffer_produce.empty(); });
                    // 交换生产和消费缓冲区
                    _fs_buffer_consume.swap(_fs_buffer_produce);
                }
                if (_async_type == AsyncType::ASYNC_SAFE)
                {
                    // 唤醒生产者
                    _fs_resource._cond_produce.notify_all();
                }
                // 调用日志落地回调函数
                _fs_callback(_fs_buffer_consume);
                // 把_buffer_consume重置
                _fs_buffer_consume.reset();
            }
        }

        void db_thread_routine()
        {
            while (true)
            {
                {
                    std::unique_lock<std::mutex> ulock(_db_resource._mutex);
                    // 当停止运行，且生产缓冲区没有数据了，才能退出
                    if (!_isrunning && _db_buffer_produce.empty())
                        break;
                    _db_resource._cond_consume.wait(ulock, [this]() -> bool
                                       { return !_isrunning || !_db_buffer_produce.empty(); });
                    // 醒了之后即使发现当停止运行，且生产缓冲区没有数据了,也不能退, 因为停止时业务线程有可能还需要放数据
                    // 因为二者想要醒来需要竞争锁, 因此不敢说当前生产缓冲区无数据我就退出,不行,需要等到下次循环时的判断

                    // 交换生产和消费缓冲区
                    _db_buffer_consume.swap(_db_buffer_produce);
                }
                if (_async_type == AsyncType::ASYNC_SAFE)
                {
                    // 唤醒生产者
                    _db_resource._cond_produce.notify_all();
                }
                // 调用日志落地回调函数
                _db_callback(_db_buffer_consume);
                // 把_buffer_consume重置
                _db_buffer_consume.clear();
            }
        }

        std::atomic<bool> _isrunning;
        Resource _fs_resource;
        Resource _db_resource;

        AsyncFSCallback _fs_callback;
        AsyncDBCallback _db_callback;
        AsyncType _async_type;

        FSBuffer _fs_buffer_produce;
        FSBuffer _fs_buffer_consume;

        DBBuffer _db_buffer_produce;
        DBBuffer _db_buffer_consume;

        std::thread _fs_worker;
        std::thread _db_worker;
    };
}
